#include"ThreadPool.h"

int threadPool_init(ThreadPool *threadpool, int worknum)
{
	pthread_mutex_init(&threadpool->_mLock, NULL);
	pthread_cond_init(&threadpool->_mCond, NULL);
	threadpool->_exitFlag = 0;
	_taskQueue_init(&threadpool->_taskQueue);
	_tidArr_init(&threadpool->_tidArr, worknum);
	return 0;
}
int _startWorker(ThreadPool *threadpool)
{
	for (int i = 0; i < threadpool->_tidArr._worker_num; i++) {
		pthread_create(&threadpool->_tidArr._arr[i], NULL, _threadFunc,threadpool);
	}
	return 0;
}

void *_threadFunc(void *arg)
{
    ThreadPool * pthreadpool = (ThreadPool *)arg;
    user_context user;
    _user_init(&user);
    printf("WORK THREAD START!\n");
    while(1)
    {
        bool beginTask = false;
        pthread_mutex_lock(&pthreadpool->_mLock);
        while(pthreadpool->_exitFlag == 0 && _taskQueue_is_empty(&pthreadpool->_taskQueue))
        {
		    pthread_cond_wait(&pthreadpool->_mCond, &pthreadpool->_mLock);
		    printf("flag = %d\n", pthreadpool->_exitFlag);
	    }
        //如果线程池终止
        if(pthreadpool->_exitFlag == 1)
        {
		    printf("recv quit from parent proccess, worker exit.\n");
            pthread_mutex_unlock(&pthreadpool->_mLock);
            pthread_exit(NULL);
	    }
	    task_t *newTask = _taskQueue_front(&pthreadpool->_taskQueue);
	    _taskQueue_deque(&pthreadpool->_taskQueue);
        beginTask = newTask && newTask->_clientFd == user._clientFd;
	    printf("worker: netfd = %d, tid = %lu\n", newTask->_clientFd,pthread_self());
	    pthread_mutex_unlock(&pthreadpool->_mLock);
	    //任务的开始
        if(beginTask)
        {
            doTask(newTask, &user);
            free(newTask);
	    }
    }
    pthread_exit(NULL);
}

int tcp_init(const char *ip, const char *port)
{
    int sockfd = socket(AF_INET,SOCK_STREAM,0);
    ERROR_CHECK(sockfd,-1,"socket create error");
    
    int flag = 1;
    setsockopt(sockfd,SOL_SOCKET,SO_REUSEADDR,&flag,sizeof(int));

    struct sockaddr_in sockAddr;
    sockAddr.sin_family = AF_INET;
    sockAddr.sin_addr.s_addr = inet_addr(ip);
    sockAddr.sin_port = htons(atoi(port));
    
    int bind_res = bind(sockfd,(struct sockaddr*)&sockAddr,sizeof(sockAddr));
    ERROR_CHECK(bind_res,-1,"socket bind error");
    
    int listen_res = listen(sockfd,SOMAXCONN);
    ERROR_CHECK(listen_res,-1,"socket listen error");

    return sockfd;
}

int epollAdd(int epfd,int fd)
{
    struct epoll_event ev;
    ev.data.fd = fd;
    ev.events = EPOLLIN;
    int epoll_ctl_res = epoll_ctl(epfd,EPOLL_CTL_ADD,fd,&ev);
    ERROR_CHECK(epoll_ctl_res,-1,"epoll add error");
    return 0;
}

int epollDel(int epfd, int fd)
{
    int epoll_ctl_res = epoll_ctl(epfd,EPOLL_CTL_DEL,fd,NULL);
    ERROR_CHECK(epoll_ctl_res,-1,"epoll delete error");
    return 0;
}





